package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object DEmo4Student {

  def main(args: Array[String]): Unit = {
    //1、创建Spark sql环境
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("sql")
      .config("spark.sql.shuffle.partitions", 1) //默认在集群中时200个
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    /**
     *
     * 5、统计总分排名前十学生各科的分数
     *
     * 6、统计总分大于平均分的学生
     *
     * 7、统计每科都及格的学生
     */

    //1、读取学生表
    val scores: DataFrame = spark.read
      .schema("id STRING,cid STRING ,score DOUBLE")
      .option("sep", ",")
      .csv("spark/data/score.txt")

    //缓存DF
    scores.cache()

    // 5、统计总分排名前十学生各科的分数
    val sumScoreDF: DataFrame = scores
      //按照学号分组
      .groupBy($"id")
      //计算总分
      .agg(sum($"score") as "sumScore")

    sumScoreDF.cache()


    //按照总分排序
    sumScoreDF.orderBy($"sumScore".desc)
      //取前十
      .limit(10)
      //将小表广播
      .hint("broadcast")
      //关联分数表
      .join(scores, "id")
      .select($"id", $"cid", $"score")
    //.show()

    scores
      //增加总分列
      .withColumn("sumScore", sum($"score") over (Window.partitionBy($"id")))
      //增加排名列
      .withColumn("r", dense_rank() over (Window.orderBy($"sumScore".desc)))
      //取前十
      .where($"r" <= 10)
    //.show()

    /**
     * 6、统计总分大于平均分的学生
     */

    val avgScoreDF: DataFrame = sumScoreDF.agg(avg($"sumSCore") as "avgScore")
    //获取第一行
    val sumScoreRow: Row = avgScoreDF.head()
    //取出字段的值
    val avgScore: Double = sumScoreRow.getAs[Double]("avgScore")

    //取出总分大于平均分的学生
    sumScoreDF
      .where($"sumSCore" > avgScore)
    //.show()

    //使用窗口函数实现
    sumScoreDF
      .withColumn("avgScore", avg("sumScore") over (Window.partitionBy(expr("'1'"))))
      .where($"sumScore" > $"avgScore")
    // .show()


    /**
     * 7、统计每科都及格的学生
     */

    val subjects: DataFrame = spark
      .read
      .schema("cid STRING,cname STRING,maxScore DOUBLE")
      .option("sep", ",")
      .csv("spark/data/subject.txt")

    scores
      //关联科目表获取总分
      .join(subjects.hint("broadcast"), "cid")
      //取出及格的分数
      .where($"score" >= $"maxScore" * 0.6)
      //计算每个学生及格的科目数量
      .withColumn("num", count($"id") over (Window.partitionBy($"id")))
      //取出都及格的学生
      .where($"num" === 6)
      .show(100)

  }
}
